package cn.witsky.rtp2ws.push.threadpool;

import java.util.Iterator;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.witsky.rtp2ws.Server;
import cn.witsky.rtp2ws.cache.UdpServerManager;
import cn.witsky.rtp2ws.constant.ThreadPoolConstant;
import cn.witsky.rtp2ws.domain.CallInfo;
import cn.witsky.rtp2ws.push.disruptor.EventCacheManage;
import cn.witsky.rtp2ws.util.CapcityUtil;
import cn.witsky.rtp2ws.util.DefaultThreadFactory;
import cn.witsky.rtp2ws.util.SystemClock;

public class PushManager {
	private static int maxTime = 2 * 60 * 60;
	private static Logger logger = LoggerFactory.getLogger(PushManager.class);
	private static List<CallInfo> callList = new CopyOnWriteArrayList<CallInfo>();
	private static ThreadPoolExecutor threadPool = null;
	public static int corePoolSize = 100;
	public static int maximumPoolSize = 200;
	public static int keepAliveTime = 60;
	public static TimeUnit timeUnit = TimeUnit.SECONDS;
	public static BlockingQueue<Runnable> waitQueue = null;
	private static ThreadFactory threadFactory = new DefaultThreadFactory(ThreadPoolConstant.wsPush);
	public static RejectedExecutionHandler handler = new DefaultRejectPolicy();
	public static Thread checkTimeOutThread = null;
	

	public static void initialize() {
		if (Server.useNewThread) {
			int times=CapcityUtil.getTime();
			corePoolSize = corePoolSize * times;
			maximumPoolSize = maximumPoolSize * times;
			if (maximumPoolSize > 2000) {
				maximumPoolSize = 2000;
			}
		}
		int capcity=2048;
		waitQueue = new LinkedBlockingQueue(capcity);
		threadPool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, timeUnit, waitQueue,
				threadFactory, handler);
		logger.info("初始化线程池 corePoolSize:{} maximumPoolSize:{} waitQueueSzie:{} keepAliveTime:{} timeUnit:{}",
				corePoolSize, maximumPoolSize, capcity, keepAliveTime, timeUnit);
		checkTimeOutThread = new Thread(new Runnable() {

			@Override
			public void run() {
				while (true) {
//					logger.debug("日志测试。。。。。。。。。。。。。。。。。。");
					Iterator<CallInfo> calls = callList.iterator();
					while (calls.hasNext()) {					
						CallInfo call = calls.next();
						PushManager.free(call);
					}
					try {
						TimeUnit.MILLISECONDS.sleep(100);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}

			}
		});
		checkTimeOutThread.setName("rtp2ws-Call-CheckTimeOut-Thread");
		checkTimeOutThread.start();
	}

	public static void offerCall(CallInfo call) {
		callList.add(call);
		if(Server.sendWsSync) {
//			call.getRtp(true).initWsClient();
//			call.getRtp(false).initWsClient();
		}else {
			if(Server.useNewThread) {
				call.getRtp(true).creatTask();
				call.getRtp(false).creatTask();
			}else {			
				EventCacheManage.bindDisruptor(call.getCallId());
//				call.initClient();
				logger.info("offer call:{}", call.getCallId());
			}
		}

	}

	public static void execTask(PushTask task) {
		threadPool.execute(task);
	}

	public static void free(CallInfo call) {
		try {	
			logger.debug("call:{} rtpCount:{}", call.getCallId(),call.getRtpCount());
			if (call.getRtpCount().get()<=0) {
				boolean isSuc=UdpServerManager.free(call.getCallId());
				if(isSuc) {
					callList.remove(call);
					EventCacheManage.clearCall(call.getCallId());
					call.getRtp(true).clientClose();
					call.getRtp(false).clientClose();
				}
//				call.clearRtp(true);
//				call.clearRtp(false);
			} else {
				long time=SystemClock.now() - call.getCreateTime();
				if ( time> maxTime * 1000) {
					logger.warn("call:{}超过{}秒的最大时长了，销毁会话", call.getCallId(),maxTime);
					call.resetRtpCount();
				}else {
					if(call.getReciveByteTime()>0) {
						time=SystemClock.now()-call.getReciveByteTime();
						if(time>10*1000) {
							logger.warn("call:{}超过10秒没有收到", call.getCallId());
							call.resetRtpCount();
						}
					}else {
						if(time>100*1000) {
							logger.warn("call:{}超过100秒没有收到", call.getCallId());
							call.resetRtpCount();
						}
					}
				}
			}
		}catch(Exception ex) {
			logger.error(call.getCallId()+"释放出现了异常",ex);
		}
	
	}
}
